// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: xguo

#include <cmath>

#include <gflags/gflags.h>
#include <google/protobuf/util/json_util.h>
#include <nlohmann/json.hpp>

#include "cc/coin2/app/init.h"
#include "coin2/base/config/app_config.h"
#include "coin2/base/config/args.h"
#include "coin2/base/hostname_util.h"
#include "coin2/base/log.h"
#include "coin2/exchange/base/log/health_reporter.h"
#include "coin2/exchange/base/log/latency_recorder_global.h"
#include "coin2/exchange/base/log/strategy_logger.h"
#include "coin2/exchange/factory/symbology/product.h"
#include "coin2/strategy/hooks/pnl_monitor.h"
#include "coin2/strategy/order_executor.pb.h"
#include "coin2/strategy/order_executor/mm_executor.h"
#include "coin2/strategy/strategy.h"
#include "coin2/strategy/util/health_reporter_util.h"
#include "coin2/strategy/util/reporter2.h"

DEFINE_string(driver, "", "driver config");
DEFINE_string(config, "", "strategy config");

namespace {

using coin2::exchange::base::feed::FeedUpdate;
using coin2::exchange::base::order::IOrderSubsystem;
using coin2::exchange::base::order::IProduct;
using coin2::exchange::base::order::OrderUpdate;
using coin2::exchange::base::strategy_util::StrategyLogger;
using coin2::strategy::PassiveExecutorConfig;
using coin2::strategy::StrategyReporter2;
using coin2::strategy::hooks::PnlMonitor;
using coin2::strategy::order_executor::PassiveOrderExecutor;

using namespace coin2::app;
using namespace coin2::base::config;
};  // namespace

class IPricer {
 public:
  virtual auto IsReady() const -> bool = 0;
  virtual auto IsMyFeed(const FeedUpdate&) const -> bool = 0;
  virtual auto trade_price() const -> double = 0;
  virtual void UpdateByBookFeed(const FeedUpdate&) = 0;
  virtual auto CalcPrice(const FeedUpdate& upd, double*, double*) -> bool = 0;
  virtual ~IPricer() = default;
};

// used by Spot/Futures
class FuturesPricer : public IPricer {
 public:
  FuturesPricer(
      int64_t window_size,
      const MarketExchangeApi& mea,
      const std::string& trade_symbol,
      const std::string& ref_symbol,
      double edge)
      : window_size_(window_size),
        mea_(mea),
        trade_symbol_(trade_symbol),
        ref_symbol_(ref_symbol),
        edge_(edge) {}

  auto IsReady() const -> bool override { return total_updates_ > 200; }

  auto IsMyFeed(const FeedUpdate& upd) const -> bool override {
    const auto& native_symbol = upd.product().native_symbol();
    return native_symbol == ref_symbol_ || native_symbol == trade_symbol_;
  }

  auto trade_price() const -> double override { return trade_price_; }

  void UpdateByBookFeed(const FeedUpdate& upd) override {
    const auto book = upd.GetBookBuilder();
    int64_t timestamp = book->Timestamp();

    if (upd.exchange() == mea_.exchange && upd.market() == mea_.market) {
      trade_price_ = 0.5 * (book->Ask0()->price + book->Bid0()->price);
      trade_count_++;
    } else {
      ref_price_ = 0.5 * (book->Ask0()->price + book->Bid0()->price);
      ref_count_++;
    }

    if (ref_count_ > 10 && trade_count_ > 10) {
      Update(std::log(trade_price_ / ref_price_), timestamp);
    }
  }

  auto CalcPrice(const FeedUpdate& upd, double* askp, double* bidp) -> bool override {
    if (upd.exchange() == mea_.exchange && upd.market() == mea_.market) {
      return false;
    }

    double avg = GetAverage();
    const auto book = upd.GetBookBuilder();
    const double ref_price = 0.5 * (book->Ask0()->price + book->Bid0()->price);
    *askp = std::exp(avg + edge_) * ref_price;
    *bidp = std::exp(avg - edge_) * ref_price;
    return true;
  }

 private:
  void Update(double x, int64_t ts) {
    data_.emplace_back(std::make_pair(ts, x));
    sum_ += x;
    while (!data_.empty()) {
      if (ts - data_.front().first > window_size_) {
        sum_ -= data_.front().second;
        data_.pop_front();
      } else {
        break;
      }
    }
    total_updates_++;
  }

  double GetAverage() const { return sum_ / data_.size(); }

 private:
  std::deque<std::pair<int64_t, double>> data_;

  const int64_t window_size_{300'000'000'000LL};
  const MarketExchangeApi mea_;
  const std::string trade_symbol_;
  const std::string ref_symbol_;
  const double edge_;

  double ref_price_{0};
  double trade_price_{0};

  int64_t ref_count_{0};
  int64_t trade_count_{0};

  int64_t total_updates_{0};
  double sum_{0};
};

// TODO:
class OptionsPricer : public IPricer {
 public:
  OptionsPricer(
      int64_t window_size,
      const MarketExchangeApi& mea,
      const std::string& trade_symbol,
      const std::string& ref_symbol,
      double edge)
      : window_size_(window_size),
        mea_(mea),
        trade_symbol_(trade_symbol),
        ref_symbol_(ref_symbol),
        edge_(edge) {}
  auto IsReady() const -> bool override { return total_updates_ > 5; }

  auto IsMyFeed(const FeedUpdate& upd) const -> bool override {
    const auto& native_symbol = upd.product().native_symbol();
    return native_symbol == trade_symbol_;
  }

  auto trade_price() const -> double override { return trade_price_; }

  void UpdateByBookFeed(const FeedUpdate& upd) override {
    const auto book = upd.GetBookBuilder();

    if (upd.exchange() != mea_.exchange || upd.market() != mea_.market) {
      return;
    }

    auto ask0 = book->Ask0();
    auto bid0 = book->Bid0();
    if (!ask0 || !bid0) {
      return;
    }

    trade_price_ = 0.5 * (ask0->price + bid0->price);
    ++total_updates_;
  }

  auto CalcPrice(const FeedUpdate& upd, double* askp, double* bidp) -> bool override {
    if (upd.exchange() != mea_.exchange || upd.market() != mea_.market) {
      return false;
    }

    const auto book = upd.GetBookBuilder();
    auto ask0 = book->Ask0();
    auto bid0 = book->Bid0();
    if (!ask0 || !bid0) {
      return false;
    }

    const double ref_price = 0.5 * (ask0->price + bid0->price);
    *askp = std::exp(1.0 + edge_) * ref_price;
    *bidp = std::exp(1.0 - edge_) * ref_price;

    if (*askp < ask0->price) {
      *askp = ask0->price;
      // if (auto ask1 = book->AskN(1); ask1) {
      //   *askp = ask1->price;
      // } else {
      //   *askp = 0;
      // }
    }
    if (*bidp > bid0->price) {
      *bidp = bid0->price;
      // if (auto bid1 = book->BidN(1); bid1) {
      //   *bidp = bid1->price;
      // } else {
      //   *bidp = 0;
      // }
    }

    return *askp > 0 || *bidp > 0;
  }

 private:
  const int64_t window_size_{300'000'000'000LL};
  const MarketExchangeApi mea_;
  const std::string trade_symbol_;
  const std::string ref_symbol_;
  const double edge_;

  int64_t total_updates_{0};
  double trade_price_{0};
};

class ProductSimpleMmStrategy : public coin2::strategy::IStrategy {
  using LatencyRecorderGlobal = coin2::exchange::base::strategy_util::LatencyRecorderGlobal;

 public:
  ProductSimpleMmStrategy(
      const nlohmann::json& strat_config,
      const PassiveExecutorConfig& exe_config)
      : strat_config_(strat_config), exe_config_(exe_config) {}

  void Init(Driver* driver) override {
    int64_t ts = GetCurrentTimestamp();
    driver_ = driver;
    auto get_product_holder =
        [this](const coin2::exchange::base::symbology::IProduct& product, int64_t timestamp) {
          return driver_->product_holder_manager()->GetProductHolder(product, timestamp);
        };
    // TODO(xguo): fix timestamp thing when you need sim
    executor_ =
        std::make_unique<PassiveOrderExecutor>(exe_config_, os(), ts, nullptr, get_product_holder);
    int64_t window_size = strat_config_["basis_ma_window"].get<int64_t>() * 1'000'000'000LL;
    trade_symbol_ =
        CreateProductFromUniqueString(strat_config_["trade_symbol"].get<std::string>(), ts);
    ref_symbol_ = CreateProductFromUniqueString(strat_config_["ref_symbol"].get<std::string>(), ts);
    double edge = strat_config_["edge_bp"].get<double>() * 1e-4;
    if (mea().market == MarketType::Options) {
      pricer_ = std::make_unique<OptionsPricer>(
          window_size,
          mea(),
          trade_symbol_->native_symbol(),
          ref_symbol_->native_symbol(),
          edge);
    } else {
      pricer_ = std::make_unique<FuturesPricer>(
          window_size,
          mea(),
          trade_symbol_->native_symbol(),
          ref_symbol_->native_symbol(),
          edge);
    }
    auto mea = this->mea();
    acc_request_.set_market_type(coin::proto::MarketType_Name(mea.market));
    acc_request_.set_exchange(coin::proto::ExchangeType_Name(mea.exchange));
    acc_request_.set_api_version(mea.api);
  }

  MarketExchangeApi mea() { return MarketExchangeApi::FromString(exe_config_.mea()); }

  IOrderSubsystem* os() { return driver_->order()->GetUniqueSystem(mea()); }

  double trade_price() const { return pricer_->trade_price(); }

  IProduct* trade_symbol() const { return trade_symbol_.get(); }

  const IProduct* GetTargetProduct() const { return trade_symbol_.get(); }
  const PassiveExecutorConfig& GetExeConfig() const { return exe_config_; }

  void onBookFeed(const FeedUpdate& upd) override {
    if (!pricer_->IsMyFeed(upd)) {
      return;
    }

    pricer_->UpdateByBookFeed(upd);
    if (!pricer_->IsReady()) {
      return;
    }

    double askp = 0.0;
    double bidp = 0.0;
    bool success = pricer_->CalcPrice(upd, &askp, &bidp);
    if (!success) {
      return;
    }

    if (is_readonly_og_) {
      LOG_FIRST_N(INFO, 1) << "ignore submit order for readonly OG";
      return;
    }
    if (os()->is_ready()) {
      LOG_FIRST_N(INFO, 3) << "os is ready!!!!";
      executor_->ManageMmOrders(upd.timestamp(), askp, bidp, 0, 0, 0, 0);
    } else {
      LOG(INFO) << "OG not ready!";
    }

#ifdef ENABLE_LATENCY_PROFILE_DEBUG_MARK
    int64_t context_id = 0;
    LatencyRecorderGlobal::SetLatencyProfileMark(
        acc_request_,
        "RAW_FEED_EXCHANGE_PUBLISHED",
        upd.ts().exchange_publish_timestamp(),
        &context_id,
        false);
    LatencyRecorderGlobal::SetLatencyProfileMark(
        acc_request_,
        "RAW_FEED_RECEIVED",
        upd.ts().raw_rx_timestamp(),
        &context_id,
        false);
    LatencyRecorderGlobal::SetLatencyProfileMark(
        acc_request_,
        "ON_BOOK_FEED",
        GetCurrentTimestamp(),
        &context_id,
        false);
#endif
  }

  void onAccountInfo(const OrderUpdate& upd) override { executor_->onAccountInfo(upd); }

  void onCleanupAux(Driver* driver, int num_retries) {
    int num_orders = 0;
    if (executor_) {
      executor_->SetEnableOrderSubmission(false);
      auto ts = GetCurrentTimestamp();
      executor_->TryCancelAllWorkingOrders(ts);
      num_orders += [&]() -> int {
        int ret = 0;
        const auto working_orders = os()->gateway()->order_manager().GetAllProductWorkingOrders(
            trade_symbol_->order_native_symbol());

        for (const auto* order : working_orders) {
          if (order->native_product() != trade_symbol_->order_native_symbol()) {
            continue;
          }
          ++ret;
        }
        return ret;
      }();
    }

    if (num_orders == 0) {
      LOG(INFO) << "Cleared all pending orders.";
      driver->PostTask([driver]() { driver->Exit(); });
    } else if (num_retries > 10) {
      LOG(WARNING) << "There may be some pendings orders left open.";
      driver->PostTask([driver]() { driver->Exit(); });
    } else {
      driver->PostDelayedTask(
          [this, driver, num_retries]() { this->onCleanupAux(driver, num_retries + 1); },
          500'000'000LL);
    }
  }

  void SetReadonlyOG(bool is_readonly_og) { is_readonly_og_ = is_readonly_og; }

 private:
  const nlohmann::json& strat_config_;
  const PassiveExecutorConfig exe_config_;
  std::unique_ptr<PassiveOrderExecutor> executor_;
  std::unique_ptr<StrategyLogger> logger_;
  std::unique_ptr<IPricer> pricer_;
  std::unique_ptr<IProduct> trade_symbol_;
  std::unique_ptr<IProduct> ref_symbol_;
  Driver* driver_;
  bool is_readonly_og_ = false;
  coin::proto::AccountRequestProto acc_request_;
};

class SimpleMmStrategy : public coin2::strategy::IStrategy {
  using LatencyRecorderGlobal = coin2::exchange::base::strategy_util::LatencyRecorderGlobal;

 public:
  explicit SimpleMmStrategy(const std::string& mm_strat_json) {
    // mm strat config
    std::ifstream in(mm_strat_json);
    LOG(INFO) << "Load strategy json: " << mm_strat_json;
    CHECK_THROW(in.is_open());
    strat_config_ = ::nlohmann::json::parse(in);

    const int64_t NS = 1'000'000'000LL;
    const std::string mea = strat_config_["mea"].get<std::string>();
    const int64_t ts = GetCurrentTimestamp();
    for (auto& [base, config] : strat_config_["products"].items()) {
      LOG(INFO) << "strategy config: " << config.dump(4);
      auto trade_symbol =
          CreateProductFromUniqueString(config["trade_symbol"].get<std::string>(), ts);
      LOG(INFO) << "Add " << trade_symbol->symbol() << " for trading." << base;
      PassiveExecutorConfig exe_config;
      exe_config.set_mea(mea);
      exe_config.set_symbol(trade_symbol->symbol());
      exe_config.set_lot_size(config["lot_size"].get<double>());
      exe_config.set_min_pos(config["min_pos"].get<double>());
      exe_config.set_max_pos(config["max_pos"].get<double>());
      exe_config.set_order_update_period(config["order_update_period_sec"].get<int64_t>() * NS);
      exe_config.set_max_posting_period(config["max_posting_period_sec"].get<int64_t>() * NS);
      if (config.find("max_working_cnt") != config.end()) {
        exe_config.set_max_working_cnt(config["max_working_cnt"].get<int>());
      }
      if (config.find("leverage") != config.end()) {
        exe_config.set_leverage(config["leverage"].get<double>());
      }
      if (config.find("allow_amend") != config.end()) {
        exe_config.set_allow_amend(config["allow_amend"].get<bool>());
      }
      if (config.find("reduce_only") != config.end()) {
        exe_config.set_reduce_only(config["reduce_only"].get<bool>());
      }
      if (config.find("reserve") != config.end()) {
        exe_config.set_reserve(config["reserve"].get<double>());
      }
      if (config.find("filter_tag") != config.end()) {
        exe_config.set_filter_tag(config["filter_tag"].get<std::string>());
      }
      exe_config.set_use_fill_pos(false);
      reserve_map_[trade_symbol->base().currency()] =
          (exe_config.max_pos() + exe_config.min_pos()) * 0.5;
      mm_strats_.push_back(std::make_unique<ProductSimpleMmStrategy>(config, exe_config));
    }
    strategy_created_time_ = ts;
  }

  MarketExchangeApi mea() {
    std::string mea = strat_config_["mea"].get<std::string>();
    return MarketExchangeApi::FromString(mea);
  }

  IOrderSubsystem* os() { return driver_->order()->GetUniqueSystem(mea()); }

  void Init(Driver* driver) override {
    driver_ = driver;

    auto interval = driver->strategy()->config().pnl_report_period_ns();
    if (interval > 0) {
      pnl_monitor_ = std::make_unique<PnlMonitor>(
          driver->live_executor(),
          driver->archive_executor(),
          interval);
      pnl_monitor_->StartAsync();  // TODO(donggu): change pnl_monitor hook to StrategySystem
    }

    coin::proto::StrategyRequestProto strat_request;
    strat_request.set_strategy_group(driver->strategy()->config().strategy_group());
    strat_request.set_strategy_name(driver->strategy()->config().strategy_name());
    // TODO(taekwon): implement 1) reserve map, 2) pnl balance later.
    LOG(INFO) << "Set Latency Recorder";
    strat_reporter_.reset(new StrategyReporter2(driver->strategy()->config()));
    for (auto& strat : mm_strats_) {
      strat->Init(driver);
      const auto& request = strat->os()->GetAccountRequest();
      const auto& exe_config = strat->GetExeConfig();
      const IProduct* product = strat->GetTargetProduct();
      strat_reporter_
          ->RegisterExecutorConfig(strategy_created_time_, request, *product, exe_config);
    }

    auto* latency_recorder =
        strat_reporter_->GetMutableStrategyLogger()->mutable_latency_recorder();
    os()->gateway()->set_latency_recorder(latency_recorder);
    LatencyRecorderGlobal::Set(latency_recorder);

    health_reporter_ = strat_reporter_->GetMutableHealthReporter();
    using RequestStatus = coin2::base::request::RequestStatus;
    auto prom_report_func = [this, strategy_name = driver->strategy()->config().strategy_name()](
                                RequestStatus infomap) -> void {
      std::vector<std::pair<prometheus::Labels, double>> res;
      LOG(INFO) << "GetApiEndPointStats: " << infomap.size();
      for (auto& [key, info] : infomap) {
        auto& [method, endpoint] = key;
        for (auto& [code, cnt] : info.http_code) {
          prometheus::Labels labels;
          labels["strategy_name"] = strategy_name;
          labels["method"] = method;
          labels["endpoint"] = endpoint;
          labels["http_code"] = std::to_string(code);
          res.emplace_back(std::make_pair(std::move(labels), cnt));
        }
      }
      if (health_reporter_) {
        health_reporter_->Write(std::move(res));
      }
    };
    os()->RegisterHealthReporter(prom_report_func);
  }

  void onBookFeed(const FeedUpdate& upd) override {
    for (auto&& strat : mm_strats_) {
      strat->onBookFeed(upd);
    }

    const auto book = upd.GetBookBuilder();
    int64_t timestamp = book->Timestamp();
    if (timestamp - last_print_ts_ > 10'000'000'000) {
      LOG(INFO) << os()->gateway()->order_manager().DebugString();
      last_print_ts_ = timestamp;
    }
  }

  void onAccountInfo(const OrderUpdate& upd) override {
    for (auto&& strat : mm_strats_) {
      strat->onAccountInfo(upd);
    }
    if (strat_reporter_) {
      for (auto& strat : mm_strats_) {
        double midp = strat->trade_price();
        if (midp < 1e-9) {
          LOG(INFO) << "midp not ready! " << midp;
        } else {
          strat_reporter_->UpdateMidp(*strat->trade_symbol(), strat->trade_price());
        }
      }
      strat_reporter_->onAccountInfo(upd);
    }
  }

  void onAccountOrder(const OrderUpdate& upd) override {
    LOG(INFO) << "receive product_order info!" << upd.product_order_info().event().DebugString();
    LOG(INFO) << "OrderContext: " << upd.product_order_info().order_contexts().size();
  }

  void onOrderLog(const coin::proto::OrderGatewayLog& og_log) override {
    if (strat_reporter_) {
      strat_reporter_->GetMutableStrategyLogger()->WriteOrderLog(
          og_log,
          true,
          GetCurrentTimestamp());
    }
  }

  void onCleanupAux(Driver* driver, int num_retries) {
    for (auto&& strat : mm_strats_) {
      strat->onCleanupAux(driver, num_retries);
    }
  }

  void onCleanup(Driver* driver) {
    driver->SetImmediateExitOnSignal(false);
    driver->PostTask([this, driver]() { this->onCleanupAux(driver, 0); });
  }

  void SetReadonlyOG(bool is_readonly_og) {
    for (auto&& strat : mm_strats_) {
      strat->SetReadonlyOG(is_readonly_og);
    }
  }

 private:
  coin2::exchange::base::strategy_util::PrometheusHealthReporter* health_reporter_;
  std::vector<std::unique_ptr<ProductSimpleMmStrategy>> mm_strats_;
  std::unique_ptr<PnlMonitor> pnl_monitor_;
  std::unique_ptr<StrategyReporter2> strat_reporter_;
  Driver* driver_{nullptr};
  int64_t last_print_ts_{0};
  nlohmann::json strat_config_;
  std::unordered_map<std::string, double> reserve_map_;
  int64_t strategy_created_time_;
};

int main(int argc, char* argv[]) {
  coin2::app::InitApplication(&argc, &argv);

  auto path = std::experimental::filesystem::path(FLAGS_driver);
  DriverConfig app_config;
  app_config.MergeFrom(coin2::base::config::DriverConfigFromJson(path));
  LOG(INFO) << "app_config: " << app_config.DebugString();

  SimpleMmStrategy strategy(FLAGS_config);
  Driver driver(app_config, &strategy);

  bool is_readonly_og = false;
  for (const auto& [name, exch] : driver.config().order().exchanges()) {
    if (!exch.has_order_logger_config()) {
      continue;
    }
    if (!exch.order_logger_config().as_account_logger()) {
      continue;
    }
    is_readonly_og = true;
    break;
  }
  if (is_readonly_og) {
    strategy.SetReadonlyOG(true);
  } else {
    driver.RegisterCleanupHandler();
  }

  driver.Run();

  return 0;
}
